package com.parking.bang.rpc.core.server;

import java.util.Map;

import com.parking.bang.rpc.serialize.base.CodecUtil;
import com.parking.bang.rpc.serialize.base.RPCSerializeProtocol;
import com.parking.bang.rpc.serialize.base.RPCSerializeSelector;
import com.parking.bang.rpc.serialize.kryo.KryoCodecUtil;
import com.parking.bang.rpc.serialize.kryo.KryoDecoder;
import com.parking.bang.rpc.serialize.kryo.KryoEncoder;
import com.parking.bang.rpc.serialize.kryo.KryoPoolFactory;

import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

/**
 * Server Serializer Selector
 * 
 * @author ivan
 *
 */
public class RPCServerSerializeSelector implements RPCSerializeSelector {

	private Map<String, Object> handlerMap = null;

	public RPCServerSerializeSelector(Map<String, Object> handlerMap) {
		this.handlerMap = handlerMap;
	}

	/**
	 * with ENUM switch, for better, use configuration and IOC
	 */
	public void select(RPCSerializeProtocol protocol, ChannelPipeline pipeline) {
		switch (protocol) {
		case JDK: {
			// sticky
			pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, CodecUtil.MESSAGE_LENGTH, 0,
					CodecUtil.MESSAGE_LENGTH));
			// refill
			pipeline.addLast(new LengthFieldPrepender(CodecUtil.MESSAGE_LENGTH));
			// encoder
			pipeline.addLast(new ObjectEncoder());
			// decoder with cache, option: disabled
			pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
					ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
			pipeline.addLast(new MessageRecvHandler(handlerMap));
			break;
		}
		case KRYO: {
			KryoCodecUtil util = new KryoCodecUtil(KryoPoolFactory.getKryoPoolInstance());
			pipeline.addLast(new KryoEncoder(util));
			pipeline.addLast(new KryoDecoder(util));
			pipeline.addLast(new MessageRecvHandler(handlerMap));
			break;
		}
		default: {
			// sticky
			pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, CodecUtil.MESSAGE_LENGTH, 0,
					CodecUtil.MESSAGE_LENGTH));
			// refill
			pipeline.addLast(new LengthFieldPrepender(CodecUtil.MESSAGE_LENGTH));
			// encoder
			pipeline.addLast(new ObjectEncoder());
			// decoder with cache, option: disabled
			pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
					ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
			pipeline.addLast(new MessageRecvHandler(handlerMap));
			break;
		}
		}
	}
}
